package com.gaofans.nrpc.server.handler;

import com.gaofans.nrpc.common.context.ContextRequest;
import com.gaofans.nrpc.common.filter.DefaultFilterChain;
import com.gaofans.nrpc.common.filter.FilterChain;
import com.gaofans.nrpc.common.processor.RpcContext;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;

/**
 * request入站过滤器
 * @author GaoFans
 */
public class RequestInFilterHandler extends SimpleChannelInboundHandler<ContextRequest> {

    private final List<ServerFilter> filters;
    private final static Logger LOGGER = LoggerFactory.getLogger(RequestInFilterHandler.class);

    public RequestInFilterHandler(List<ServerFilter> filters) {
        this.filters = Objects.requireNonNull(filters);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ContextRequest request) {
        if (filters.size() > 0){
            request.executor().submit(() -> {
                RpcContext context = new RpcContext(ctx, request.getSequenceId());
                FilterChain<ContextRequest> chain = new DefaultFilterChain<>(filters,context);
                try {
                    chain.doFilter(request)
                            .doOnSuccess(unused -> ctx.fireChannelRead(request))
                            .doOnError(throwable -> {
                                context.clear();
                                LOGGER.error("过滤器链调用异常", throwable);
                            })
                            .subscribe();
                }catch (Exception e){
                    context.clear();
                }
            });
        }else {
            ctx.fireChannelRead(request);
        }
    }
}
